package com.my;

import com.alibaba.fastjson.JSONObject;
import com.my.serialize.MySqlDeserialization;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;


public class App2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //源算子
        MySqlSource<String> mysqlSource = MySqlSource.<String>builder()
                .hostname("192.168.80.101")
                .port(13306)
                .username("root")
                .password("root")
                .databaseList("test")
                .tableList("test.*")
                //反序列化策略
                .deserializer(new MySqlDeserialization())
                //起始位置为从最新开始消费
                .startupOptions(StartupOptions.latest())
//                .startupOptions(StartupOptions.initial())
                .serverTimeZone("Asia/Shanghai")
                .includeSchemaChanges(true)
                .build();
        DataStreamSource<String> streamSource = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "mysqlSource");
        //输出算子

        streamSource.addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value) throws Exception {
                System.out.println(value);
            }
        }).name("sink_es");

        env.execute("mysql_es");
    }
}
